package org.big.kafkafinancesimulator.simulator;

import org.big.kafkafinancesimulator.model.TradeRequest;
import org.big.kafkafinancesimulator.model.TradeResult;
import org.big.kafkafinancesimulator.model.UserAccount;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Service
public class TradingProcessorService {
    private static final String TOPIC_TRADE_REQUESTS = "trade-requests";
    private static final String TOPIC_MARKET_DATA = "market-data";
    private static final String TOPIC_TRADE_RESULTS = "trade-results";

    @Autowired

    private KafkaTemplate<String, String> kafkaTemplate;

    // 使用ConcurrentHashMap在内存中维护用户状态
    private final Map<String, UserAccount> userAccountMap = new ConcurrentHashMap<>();
    private final Map<String, Map<String, Integer>> userPortfolioMap = new ConcurrentHashMap<>(); // Key: userId, Value: Map<Symbol, Quantity>

    @KafkaListener(topics = TOPIC_TRADE_REQUESTS, groupId = "trading-group")
    public void processTradeRequest(TradeRequest request) {
        TradeResult result = new TradeResult();
        result.setRequestId(request.getRequestId());
        result.setUserId(request.getUserId());
        result.setType(request.getType());
        result.setSymbol(request.getSymbol());
        result.setQuantity(request.getQuantity());

        // 1. 获取或创建用户账户和持仓
        UserAccount account = userAccountMap.computeIfAbsent(request.getUserId(), k -> new UserAccount(k, 100000.00)); // 初始资金10万
        Map<String, Integer> portfolio = userPortfolioMap.computeIfAbsent(request.getUserId(), k -> new ConcurrentHashMap<>());

        // 2. 确定执行价格（限价单用请求价，市价单用最新市价）
        double executionPrice = (request.getPrice() != null) ? request.getPrice() : getCurrentMarketPrice(request.getSymbol());
        result.setExecutedPrice(executionPrice);

        double totalCost = executionPrice * request.getQuantity();

        try {
            if (request.getType() == TradeRequest.Type.BUY) {
                // 买入逻辑：检查资金是否足够
                if (account.getBalance() >= totalCost) {
                    account.setBalance(account.getBalance() - totalCost);
                    portfolio.merge(request.getSymbol(), request.getQuantity(), Integer::sum);
                    result.setStatus(TradeResult.Status.SUCCESS);
                    result.setMessage("买入成功");
                } else {
                    result.setStatus(TradeResult.Status.FAILED);
                    result.setMessage("资金不足");
                }
            } else { // SELL
                // 卖出逻辑：检查持仓是否足够
                int currentHolding = portfolio.getOrDefault(request.getSymbol(), 0);
                if (currentHolding >= request.getQuantity()) {
                    account.setBalance(account.getBalance() + totalCost);
                    portfolio.put(request.getSymbol(), currentHolding - request.getQuantity());
                    result.setStatus(TradeResult.Status.SUCCESS);
                    result.setMessage("卖出成功");
                } else {
                    result.setStatus(TradeResult.Status.FAILED);
                    result.setMessage("持仓不足");
                }
            }
            result.setTotalAmount(totalCost);
        } catch (Exception e) {
            result.setStatus(TradeResult.Status.FAILED);
            result.setMessage("处理交易时发生错误: " + e.getMessage());
        }

        // 3. 将处理结果发送到另一个Kafka Topic，可供前端（如WebSocket）或其它系统消费
        kafkaTemplate.send(TOPIC_TRADE_RESULTS, request.getUserId(), result.toString());
    }

    // 模拟获取当前市场价格（实际应从本地缓存或最新Kafka消息中获取）
    private double getCurrentMarketPrice(String symbol) {
        // 这里可以注入MarketDataSimulatorService来获取价格
        // 简化起见，返回一个模拟值
        return 150.0;
    }
}